#include "DialServicesSDK.h"
#include <stdlib.h>
#include <string.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/types.h>
#include <pthread.h>

static int iDialDebugLog = 1;
static void dialServicer_on_connect(struct mosquitto *mosq, void *obj, int rc);
static void dialServicer_on_disconnect(struct mosquitto *mosq, void *obj, int rc);
static void dialServicer_on_message_v5(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg, const mosquitto_property *props);
static void dialClient_on_message_v5(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg, const mosquitto_property *props);
static void dialClient_on_connect(struct mosquitto *mosq, void *obj, int rc);
static void dialClient_on_disconnect(struct mosquitto *mosq, void *obj, int rc);

ABC GG
FF

FFF

CDialServicerAPI::CDialServicerAPI()
{
	m_pMosq = NULL;
	m_strBrokerAddr = "";
	m_userServFun = NULL;
	m_userServResp = NULL;
	m_qos = 0;

	mosquitto_lib_init();
}

CDialServicerAPI::~CDialServicerAPI()
{
	mosquitto_destroy(m_pMosq);
	mosquitto_lib_cleanup();
}


int CDialServicerAPI::ServiceSet(const string servName, ServicsFunc userServFun, ServicsResp userServResp
									, const string brokerIp, const int brokerPort)
{
	const char *cstr;
	int ver = PROTOCOL_VERSION_v5;
	void *userdata = (void *)this;

	
	if (servName.empty() || servName.size() > CN_DIALNAME_MAXLEN ||userServFun == NULL || userServResp == NULL)
		return CN_DIALERRNO_PARAMERR;

	m_servName = servName;
	cstr = m_servName.c_str();

	m_userServFun = userServFun;
	m_userServResp = userServResp;

	if (m_pMosq != NULL)
	{
	    mosquitto_destroy(m_pMosq);
		m_pMosq = NULL;
	}
	m_pMosq = mosquitto_new(cstr, true, userdata);
	if(m_pMosq == NULL)
	    return CN_DIALERRNO_UNABLEALLOC;
	
	mosquitto_opts_set(m_pMosq, MOSQ_OPT_PROTOCOL_VERSION, &ver);
	
	mosquitto_connect_callback_set(m_pMosq, dialServicer_on_connect);
	mosquitto_disconnect_callback_set(m_pMosq, dialServicer_on_disconnect);
	mosquitto_message_v5_callback_set(m_pMosq, dialServicer_on_message_v5);
	
    m_strBrokerAddr = brokerIp;
	m_brokerPort = brokerPort;

    return CN_DIALERRNO_SUCCESS;
}


int CDialServicerAPI::ServiceStart(const string servName, ServicsFunc userServFun, ServicsResp userServResp
	, string brokerIp, int brokerPort)
{
	int iRet;
	
	iRet = ServiceSet(servName, userServFun, userServResp, brokerIp, brokerPort);
    if (iRet != CN_DIALERRNO_SUCCESS)
    {
		if (iDialDebugLog)
			printf("ServiceStart ServiceSet failed, iRet=%d\n", iRet);
		return iRet;
    }

	iRet = mosquitto_connect_async(m_pMosq, m_strBrokerAddr.c_str(), m_brokerPort, 60);
    if (iRet != MOSQ_ERR_SUCCESS)
    {
        if (iDialDebugLog)
			printf("ServiceStart mosquitto_connect_async failed, iRet=%d\n", iRet);
		return CN_DIALERRNO_CONNECTERR;
    }
	if (iDialDebugLog)
		printf("ServiceStart try mosquitto_connect_async success\n");


	iRet = mosquitto_loop_start(m_pMosq);
	if (iRet != MOSQ_ERR_SUCCESS)
	{
		if (iDialDebugLog)
			printf("ServiceStart mosquitto_loop_start failed, iRet=%d\n", iRet);
		return CN_DIALERRNO_SERVINGERR;
	}
	
	if (iDialDebugLog)
		printf("ServiceStart mosquitto_loop_start success, name=%s\n", m_servName.c_str());
	
	return CN_DIALERRNO_SUCCESS;
}

string CDialServicerAPI::GetServName()
{
	return m_servName;
}

ServicsFunc CDialServicerAPI::GetServFun()
{
    return m_userServFun;
}

ServicsResp CDialServicerAPI::GetServResp()
{
    return m_userServResp;
}

string CDialServicerAPI::GetServBrokerAddr()
{
    return m_strBrokerAddr;
}


int CDialServicerAPI::GetServbrokerPort()
{
    return m_brokerPort;
}

int CDialServicerAPI::GetQos()
{
    return m_qos;
}


/*
由于MQTT大部分的应用场景是物联网，网络可能出现断网的情况．如果断网重连的话，重新连接后一定要再次订阅topic.
不然对应的消息可能会收不到．尤其是在使用mosquitto的mosquitto_loop_forever()接口时，这个接口有自动重连的机制，
开始我在使用的时候就忽略了这一点，导致自动重连的时候没有订阅topic.最好的做法是在connect的callback函数中
设置订阅topic,这样每次重连后都能自动订阅.
*/
void dialServicer_on_connect(struct mosquitto *mosq, void *obj, int rc)
{
	int nRet;
	CDialServicerAPI * pAPI = (CDialServicerAPI *) obj;

	if (rc)
	{
	    if (iDialDebugLog)
			printf("dial_on_connect erro, rc=%d\n", rc);
		return;
	}
	else
	{
		if (iDialDebugLog)
			printf("dial_on_connect, success \n");

		nRet = mosquitto_subscribe(mosq, NULL, pAPI->GetServName().c_str(), 2); //需要订阅上
		if (nRet != MOSQ_ERR_SUCCESS)
		{
			if (iDialDebugLog)
				printf("dial_on_connect, mosquitto_subscribe faild, nRet=%d, servName=%s\n", nRet, pAPI->GetServName().c_str());
			return ;
		}
        else
        {
            if (iDialDebugLog)
				printf("dial_on_connect, mosquitto_subscribe success\n");
			return ;
        }
	}
}


void dialServicer_on_disconnect(struct mosquitto *mosq, void *obj, int rc)
{
	CDialServicerAPI * pAPI = (CDialServicerAPI *) obj;
    CN_DAILUNUSED(mosq) ;
	if (rc)
	{
		if (iDialDebugLog)
			printf("service %s, dialServicer_on_disconnect erro, rc=%d\n", pAPI->GetServName().c_str(),rc);
		return;
	}
}


void dialServicer_on_message_v5(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg, const mosquitto_property *props)
{
	const mosquitto_property *p_resp = NULL;
	const mosquitto_property *p_corr = NULL;
	mosquitto_property *p_Retprops = NULL;
	char *resp_topic = NULL;
	int rc;
	ServicsFunc	userServFun;
	ServicsResp userServResp;
	char * pRespMsg;
	int iRespMsgLen;
	void *pValue;
	uint16_t len;
    unsigned int invokeID; 

	iRespMsgLen = 0;
	pRespMsg = NULL;
	invokeID = 0;
	pValue  = NULL;
	
	CDialServicerAPI * pAPI = (CDialServicerAPI *) obj;
	userServFun = pAPI->GetServFun( );
	userServResp = pAPI->GetServResp();

	if (iDialDebugLog)
		printf("dialServicer_on_message_v5 topic=%s, msgLen=%d, payload=%s\n", msg->topic, msg->payloadlen, (char * )msg->payload);
	
	if(!strncmp(msg->topic, pAPI->GetServName().c_str(), strlen(msg->topic)))
	{
	    p_resp = mosquitto_property_read_string(props, MQTT_PROP_RESPONSE_TOPIC, &resp_topic, false);
		if (iDialDebugLog)
		{
        	if (p_resp)
				printf("dialServicer_on_message_v5, RESPONSE_TOPIC=%s\n", resp_topic);
			else
			    printf("dialServicer_on_message_v5, RESPONSE_TOPIC = NULL\n");
		}
		
		if(p_resp)
		{
		    p_corr = mosquitto_property_read_binary(props, MQTT_PROP_CORRELATION_DATA, &pValue, &len, false);
			if (p_corr && len == sizeof(invokeID))
			{
				invokeID = *((unsigned int *)pValue);
				if (iDialDebugLog)
					printf("dialServicer_on_message_v5, CORRELATION_DATA, invokeID=%d\n", invokeID);
				//调用回调函数
           		if (userServFun != NULL)
            	{
					userServFun((char * )msg->payload, msg->payloadlen, &pRespMsg , &iRespMsgLen);
					if (iDialDebugLog)
               			printf("dialServicer_on_message_v5, userServFun success, iRespMsgLen=%d\n", iRespMsgLen);
					if (iRespMsgLen>0 && iRespMsgLen<=CN_DIALMSG_MAXLEN && pRespMsg != NULL)
					{
						//发送请求信息
						mosquitto_property_add_string(&p_Retprops, MQTT_PROP_RESPONSE_TOPIC, pAPI->GetServName().c_str());
						mosquitto_property_add_binary(&p_Retprops, MQTT_PROP_CORRELATION_DATA, &invokeID, sizeof(invokeID));
						rc = mosquitto_publish_v5(mosq, NULL, (const char *)resp_topic
									, iRespMsgLen, (const void *)pRespMsg, pAPI->GetQos(), false, p_Retprops);
						if (rc == MOSQ_ERR_SUCCESS)
                		{
                    		if (iDialDebugLog)
                        		printf("dialServicer_on_message_v5, mosquitto_publish_v5 success, iRespMsgLen=%d\n", iRespMsgLen);
                		}
						else
						{
				    		if (iDialDebugLog)
				        		printf("dialServicer_on_message_v5, mosquitto_publish_v5 rc\n");
						}
						mosquitto_property_free_all(&p_Retprops);
					}
            	}
				if (userServResp != NULL)
					userServResp(pRespMsg);
			}
		}
	}


	if (pValue)
	    free(pValue);
	if (resp_topic)
	    free(resp_topic);
}


CDialClientAPI::CDialClientAPI()
{
	int flags;
	m_pMosq = NULL;
	m_invokeID = 66;
	m_pRespBuf = NULL;
	m_ResponsesLen = 0;
	m_invokeIDFromMsg  = 0 ;
	m_qos = 0;

	if (pipe(m_pipFiledes) >= 0)
	{
	    //pipe 设置为非阻塞模式，支持异步调用
		flags = fcntl(m_pipFiledes[0],F_GETFL);
		if (flags >= 0)
		{
			flags |= O_NONBLOCK;
			fcntl(m_pipFiledes[0],F_SETFL,flags);
		}
		flags = fcntl(m_pipFiledes[1],F_GETFL);
		if (flags >= 0)
		{
			flags |= O_NONBLOCK;
			fcntl(m_pipFiledes[1],F_SETFL,flags);
		}
	}
	
	m_pRespBuf = (char*)malloc(CN_DIALMSG_MAXLEN);
	mosquitto_lib_init();
	pthread_mutex_init(&m_mutex, NULL);
}

CDialClientAPI::~CDialClientAPI()
{
	int nRet;
	tagRespInfo tRespInfo;
	
	nRet = CN_DIALERRNO_SUCCESS;
	//释放队列里的指针申请的内存
	while (nRet == CN_DIALERRNO_SUCCESS)
	{
		nRet = Receive(&tRespInfo);
		if (nRet == CN_DIALERRNO_SUCCESS)
			FreeRespInfo(&tRespInfo);
	}
	
	close( m_pipFiledes[0] );
	close( m_pipFiledes[1] );
	mosquitto_destroy(m_pMosq);
	mosquitto_lib_cleanup();
	pthread_mutex_destroy(&m_mutex);

}

int CDialClientAPI::ClientStart(const string clientName, const string brokerIp, const int brokerPort)
{
    int iRet;
	const char *cstr;
	int ver = PROTOCOL_VERSION_v5;
	void *userdata = (void *)this;

	
	if (clientName.empty() || clientName.size() > CN_DIALNAME_MAXLEN)
		return CN_DIALERRNO_PARAMERR;
	
	cstr = clientName.c_str();
	m_clientName = clientName;
	
	if (m_pMosq != NULL)
	{
	    mosquitto_destroy(m_pMosq);
		m_pMosq = NULL;
	}
	m_pMosq = mosquitto_new(cstr, true, userdata);
	if(m_pMosq == NULL)
	    return CN_DIALERRNO_UNABLEALLOC;
	
	mosquitto_opts_set(m_pMosq, MOSQ_OPT_PROTOCOL_VERSION, &ver);
	mosquitto_connect_callback_set(m_pMosq, dialClient_on_connect);
	mosquitto_disconnect_callback_set(m_pMosq, dialClient_on_disconnect);
	mosquitto_message_v5_callback_set(m_pMosq, dialClient_on_message_v5);
	
    m_strBrokerAddr = brokerIp;
	m_brokerPort = brokerPort;

	iRet = mosquitto_connect_async(m_pMosq, m_strBrokerAddr.c_str(), m_brokerPort, 60);
    if (iRet != MOSQ_ERR_SUCCESS)
    {
        if (iDialDebugLog)
			printf("ClientSet mosquitto_connect_async failed, iRet=%d\n", iRet);
		return CN_DIALERRNO_CONNECTERR;
    }
	if (iDialDebugLog)
		printf("ClientSet mosquitto_connect_async success\n");

	
	iRet = mosquitto_loop_start(m_pMosq);
	if (iRet != MOSQ_ERR_SUCCESS)
	{
		if (iDialDebugLog)
			printf("ServiceStart mosquitto_loop_start failed, iRet=%d\n", iRet);
		return CN_DIALERRNO_SERVINGERR;
	}
	
	if (iDialDebugLog)
		printf("ServiceStart mosquitto_loop_start success\n");

	return CN_DIALERRNO_SUCCESS;
}


int CDialClientAPI::Request(const string servName, char *request, int requestLen)
{
	int iRet;
	const char *servCstr;
	string setServName;
											
	mosquitto_property *props = NULL;
									
	if (request == NULL || requestLen > CN_DIALMSG_MAXLEN || servName.empty() || servName.size() > CN_DIALNAME_MAXLEN)
	{
		return CN_DIALERRNO_PARAMERR;
	}
									
	if (m_pMosq == NULL)
	{
	    return CN_DIALERRNO_UNINIT;
	}
																		
	servCstr = servName.c_str();
	if (iDialDebugLog)
	    printf("Calling servName=%s\n", servCstr);
	
	//发送请求信息
	mosquitto_property_add_string(&props, MQTT_PROP_RESPONSE_TOPIC, m_clientName.c_str());
	AddInvokeID();
	mosquitto_property_add_binary(&props, MQTT_PROP_CORRELATION_DATA, &m_invokeID, sizeof(m_invokeID));
	iRet = mosquitto_publish_v5(m_pMosq, NULL, servCstr, requestLen, request, m_qos, 0, props);
	if (iRet != MOSQ_ERR_SUCCESS)
	{
	    if (iDialDebugLog)
			printf("Calling mosquitto_publish_v5 failed! iRet = %d\n", iRet);
		iRet = CN_DIALERRNO_REQERR;
	}
	else
	{
		if (iDialDebugLog)
			printf("Calling mosquitto_publish_v5 success! requestLen = %d\n", requestLen);
		iRet = CN_DIALERRNO_SUCCESS;
	}
	
	mosquitto_property_free_all(&props);
    return iRet;
}


int CDialClientAPI::WaitResponse(tagRespInfo *ptRespInfo, int timeOutMs)
{
	int iRet;
	fd_set rd;
	struct timeval tv;

	if (ptRespInfo == NULL)
	{
		return CN_DIALERRNO_PARAMERR;
	}
	
    if (m_pMosq == NULL)
    {
        return CN_DIALERRNO_UNINIT;
    }

	//等待接收请求(超时等待)
	FD_ZERO(&rd);
	FD_SET(m_pipFiledes[0],&rd);

	if (timeOutMs >= 0)
	{
		tv.tv_sec = timeOutMs/1000;
		tv.tv_usec = (timeOutMs%1000)*1000;
		iRet = select(m_pipFiledes[0]+1, &rd,NULL,NULL,&tv);
	}
	else
		iRet = select(m_pipFiledes[0]+1, &rd,NULL,NULL,NULL);
	
    if (iRet < 0)
    {
        iRet = CN_DIALERRNO_RESPERR;
    }
	else if (iRet == 0)
	{
	    iRet = CN_DIALERRNO_TIMEOUT;
	}
	else
	{
		iRet = Receive(ptRespInfo);
	}
	return iRet;
}


void CDialClientAPI::FreeRespInfo(tagRespInfo * ptRespInfo)
{
	if (ptRespInfo->pRespBuf)
	{
		free(ptRespInfo->pRespBuf);
	}
}


void CDialClientAPI::AddInvokeID()
{
    pthread_mutex_lock(&m_mutex);
	m_invokeID++;
	pthread_mutex_unlock(&m_mutex);
}

int CDialClientAPI::GetInvokeID()
{
    int invokeID = 0;
    pthread_mutex_lock(&m_mutex);
	invokeID = m_invokeID;
	pthread_mutex_unlock(&m_mutex);
	return invokeID ;
}


void CDialClientAPI::SetRespBuf(char * pRespBuf , int responsesLen, unsigned int invokeIDFromMsg)
{
    int iCopyLen=0;
    pthread_mutex_lock(&m_mutex);
	iCopyLen = min(responsesLen, CN_DIALMSG_MAXLEN);
	m_ResponsesLen = iCopyLen;
	memcpy(m_pRespBuf, pRespBuf, iCopyLen);
	m_invokeIDFromMsg = invokeIDFromMsg;
	pthread_mutex_unlock(&m_mutex);
	return;
}

void CDialClientAPI::GetRespBuf(char	* pRespBuf , int *responsesLen, unsigned int *invokeIDFromMsg)
{
    pthread_mutex_lock(&m_mutex);
	memcpy(pRespBuf, m_pRespBuf, m_ResponsesLen);
	*responsesLen = m_ResponsesLen;
	*invokeIDFromMsg = 	m_invokeIDFromMsg;
	pthread_mutex_unlock(&m_mutex);
	return;
}

string CDialClientAPI::GetClientName()
{
	return m_clientName;
}

int  CDialClientAPI::GetQos()
{
	return m_qos;
}

int CDialClientAPI::Notify(tagRespInfo * pRespInfo)
{
    ssize_t  nRet;	
	tagRespInfo tRespInfo;

	tRespInfo = *pRespInfo;

	nRet = write(m_pipFiledes[1], (char*)&tRespInfo, sizeof(tagRespInfo));
	if (nRet <= 0)
	{
	    return CN_DIALERRNO_PIPEERR;
	}
	else
	{
		return CN_DIALERRNO_SUCCESS;
	}
}

int CDialClientAPI::Receive(tagRespInfo * pRespInfo)
{
    ssize_t  nRet;
	tagRespInfo tRespInfo;

	nRet = read(m_pipFiledes[0], (char *)&tRespInfo, sizeof(tagRespInfo));
	if (nRet != sizeof(tagRespInfo))
	{
		return CN_DIALERRNO_PIPEERR;
	}
	else
	{
		*pRespInfo = tRespInfo;
		return CN_DIALERRNO_SUCCESS;
	}
}



void dialClient_on_connect(struct mosquitto *mosq, void *obj, int rc)
{
	CDialClientAPI * pAPI = (CDialClientAPI *) obj;

	if (rc)
	{
	    if (iDialDebugLog)
			printf("dialClient_on_connect erro, rc=%d\n", rc);
	}
	else
	{
		if (iDialDebugLog)
			printf("dialClient_on_connect, success \n");
		rc = mosquitto_subscribe(mosq, NULL, pAPI->GetClientName().c_str(), 2);
		if (rc)
		{
			if (iDialDebugLog)
				printf("dialClient_on_connect, mosquitto_subscribe failed, rc=%d\n", rc);
		}
	}
}


void dialClient_on_disconnect(struct mosquitto *mosq, void *obj, int rc)
{
	CDialClientAPI * pAPI = (CDialClientAPI *) obj;

	CN_DAILUNUSED(mosq);

	if (iDialDebugLog)
		printf("client name %s, dialClient_on_disconnect , rc=%d\n", pAPI->GetClientName().c_str(), rc);
}

void dialClient_on_message_v5(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg
			, const mosquitto_property *props)
{
	int 	nRet;
	unsigned int 	invokeID;
	char *resp_topic = NULL;
	void *pValue = NULL;
	uint16_t len;
	const mosquitto_property  *p_corr = NULL;
	const mosquitto_property  *p_resp = NULL;
	tagRespInfo tRespInfo;
	
	CDialClientAPI * pAPI = (CDialClientAPI *) obj;

	if (mosq == NULL || obj == NULL || msg == NULL|| props == NULL)
		return ;
	
	if (iDialDebugLog)
		printf("dialClient_on_message_v5, msg->topic=%s\n", msg->topic);

	
	if(!strncmp(msg->topic, pAPI->GetClientName().c_str(), strlen(msg->topic)))
	{
	    p_resp = mosquitto_property_read_string(props, MQTT_PROP_RESPONSE_TOPIC, &resp_topic, false);
		if (iDialDebugLog)
		{
        	if (p_resp)
				printf("dialServicer_on_message_v5, RESPONSE_TOPIC=%s\n", resp_topic);
			else
			    printf("dialServicer_on_message_v5, RESPONSE_TOPIC = NULL\n");
		}
		if(p_resp)
		{			
			p_corr = mosquitto_property_read_binary(props, MQTT_PROP_CORRELATION_DATA, &pValue, &len, false);
			if (p_corr && len == sizeof(invokeID) && msg->payloadlen > 0)
			{
				tRespInfo.respLen = min(msg->payloadlen, CN_DIALMSG_MAXLEN);
				tRespInfo.pRespBuf = (char	*)malloc(tRespInfo.respLen);
				if (tRespInfo.pRespBuf)
				{
					memcpy(tRespInfo.pRespBuf, msg->payload, tRespInfo.respLen);
					strcpy(tRespInfo.servName, resp_topic);
					invokeID = *((unsigned int *)pValue);
					tRespInfo.invokeID = invokeID;
					nRet = pAPI->Notify(&tRespInfo);
					if (nRet != CN_DIALERRNO_SUCCESS)
					{
						free(tRespInfo.pRespBuf);
					}
				}
			}
		}
	}
	if (resp_topic)
	    free(resp_topic);
	if (pValue)
	    free(pValue);
}